package com.qqt.csr.ixport.timer;

import cn.hutool.core.thread.ThreadUtil;
import com.qqt.csr.common.utils.CacheKeyUtil;
import com.qqt.csr.common.utils.RedissonLockUtil;
import com.qqt.csr.common.utils.SnowflakeUtil;
import com.qqt.csr.common.utils.ThreadPoolUtil;
import com.qqt.csr.ixport.entity.BatchImportTask;
import com.qqt.csr.ixport.entity.ImportTopicConfig;
import com.qqt.csr.ixport.service.BatchImportTaskService;
import com.qqt.csr.ixport.service.FileChunkProducer;
import com.qqt.csr.ixport.service.ImportTopicConfigService;
import com.qqt.csr.ixport.task.TaskProcessor;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class ImportTaskHandlerTimer implements InitializingBean {
    private static final String LOG_TID = "ctid";
    private volatile boolean isShutdown = false;
    private final ScheduledThreadPoolExecutor scheduledExecutor = ThreadUtil.createScheduledExecutor(1);
    private final ExecutorService workerPool = ThreadPoolUtil.newThreadPoolExecutorCallerRunsRejectPolicy(3, 10, 60, 300, "import_task_timer_prc");

    @Value(value = "${import.task.timer.interval:15}")
    private long interval;
    @Value(value = "${import.task.handle.limit:300}")
    private int limit;
    @Value(value = "${import.task.handler.timer.switch:true}")
    private Boolean switchEnable;

    @Autowired
    private BatchImportTaskService batchImportTaskService;
    @Autowired
    private RedissonLockUtil redissonLockUtil;
    @Autowired
    private FileChunkProducer fileChunkProducer;
    @Autowired
    private ImportTopicConfigService importTopicConfigService;

    @Override
    public void afterPropertiesSet() throws Exception {
        ThreadUtil.schedule(scheduledExecutor,
                () -> {
                    if (!switchEnable) {
                        return;
                    }
                    try {
                        MDC.put(LOG_TID, SnowflakeUtil.nextIdStr());
                        this.lookup();
                    } catch (Throwable e) {
                        log.error("处理导入任务执行异常", e);
                    } finally {
                        MDC.clear();
                    }
                },
                30,
                interval,
                TimeUnit.SECONDS,
                false
        );
    }

    @PreDestroy
    public void destroy() {
        isShutdown = true;
        ThreadPoolUtil.shutdownThreadPoolGracefully(workerPool);
        ThreadPoolUtil.shutdownThreadPoolGracefully(scheduledExecutor);
        log.info("处理导入任务执行已关闭");
    }

    /**
     * 循环处理
     */
    private void lookup() {
        // 获取任务
        Long lastId = 0L;
        List<BatchImportTask> taskList = null;
        do {
            taskList = batchImportTaskService.fetchUnStartStatus(lastId, limit);
            if (isShutdown || CollectionUtils.isEmpty(taskList)) {
                return;
            }

            for (BatchImportTask task : taskList) {
                // 多线程处理任务
                String cacheKey = String.format(CacheKeyUtil.Ixport.PROCESS_TASK_LOCK, task.getId());
                redissonLockUtil.distributedLock(() -> {
                    workerPool.submit(createTaskProcessor(task));
                }, cacheKey, 0, 30, TimeUnit.MINUTES);
            }

            lastId = taskList.get(taskList.size() - 1).getId();
        } while (!isShutdown);
    }

    /**
     * 处理任务
     *
     * @param task
     */
    private TaskProcessor createTaskProcessor(BatchImportTask task) {
        ImportTopicConfig topicConfig = importTopicConfigService.getByBusinessCode(task.getBusinessCode());
        return new TaskProcessor(task, batchImportTaskService, fileChunkProducer, topicConfig);
    }

}
